package com.nhncorp.pleiades.ens;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jgroups.ChannelException;
import org.jgroups.ChannelFactory;
import org.jgroups.JChannel;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.blocks.ReplicatedHashMap;

import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.common.Constants;
import com.nhncorp.pleiades.conf.PleiadesConfiguration;
import com.nhncorp.pleiades.hadoop.New_RPC;
import com.nhncorp.pleiades.hadoop.New_Server;
import com.nhncorp.pleiades.msg.MasterServerInfo;
import com.nhncorp.pleiades.msg.MemberServerInfo;
import com.nhncorp.pleiades.msg.MessageParser;
import com.nhncorp.pleiades.msg.ParserFactory;
import com.nhncorp.pleiades.protocol.EventMetaData;
import com.nhncorp.pleiades.protocol.EventServiceProtocol;
import com.nhncorp.pleiades.util.NetworkUtil;
import com.nhncorp.pleiades.util.Sleeper;

public class EventManager implements EventManagerInterface {
  
  public static final Log LOG = LogFactory.getLog(EventManager.class.getName());
  private PleiadesConfiguration conf;
  private String hostName;
  private String eventChannelName;
  private JChannel eventChannel = null;
  private JChannel notificationChannel = null;
  private ReplicatedHashMap<String, LeaseTime> eventEntryStore = null;
  private ReplicatedHashMap<String, EventMetaData[]> eventSpace = null;
  private ReplicatedHashMap<String, EventMetaData[]> userSpace = null;
  private New_Server server;
  private long pollingInterval = 1 * 1000;
  private boolean isPrimaryNode = false;
  private AtomicBoolean stopRequested = new AtomicBoolean(false);
  private Sleeper sleeper;
  private MessageParser messageParser;
  private String clusterID;
  private List<Thread> runningThreads = new ArrayList<Thread>();
  private String masterEventManager;
  Queue<MemberServerInfo> membershipQueue;
  AtomicInteger numberOfEventManagers;
  private List<LockEventDeliver> eventDeliverList;
  private final LinkedList<EventMessage> eventDeliveryQueue = new LinkedList<EventMessage>();
  
  public EventManager(PleiadesConfiguration conf) {

    try {
      System.setProperty("java.net.preferIPv4Stack", "true");
      InetSocketAddress serverAddress = NetworkUtil.getAddress(InetAddress
          .getLocalHost().getHostName() + ":" + conf.getInt("eventManager.port", 10000));
      this.hostName = serverAddress.getHostName() + ":" + serverAddress.getPort();
      this.conf = conf;
      this.clusterID = conf.get("clusterID", "clusterID");
      this.eventChannelName = this.clusterID
          + conf.get("eventChannelName", Constants.EVENT_CHANNEL_NAME);
      
      membershipQueue = new ConcurrentLinkedQueue<MemberServerInfo>();
      numberOfEventManagers = new AtomicInteger(0);
      ChannelFactory lockFactory = new JChannelFactory();
      eventEntryStore = new ReplicatedHashMap<String, LeaseTime>(clusterID
          + conf.get("eventEntryStore", Constants.EVENT_ENTRY_STORE_NAME),
          lockFactory, Constants.LOCK_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);
      
      eventSpace = new ReplicatedHashMap<String, EventMetaData[]>(clusterID
          + conf.get("evnetSpace", Constants.EVENT_SPACE_NAME),
          lockFactory, Constants.LOCK_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);
      
      userSpace = new ReplicatedHashMap<String, EventMetaData[]>(clusterID
          + conf.get("evnetSpace", Constants.EVENT_SPACE_NAME),
          lockFactory, Constants.LOCK_ADDRESS + Constants.SOCK_PROPERTY, false, 1000);      
      
      messageParser = ParserFactory.newInstance(ParserFactory.PARSER_TYPE.EVENT_MANAGER);
      this.sleeper = new Sleeper(1000, this.stopRequested);
      LOG.info("====================EventManager Handler:"
          + serverAddress + "=====================");
      this.server = New_RPC.getServer(this, serverAddress.getHostName(), serverAddress.getPort(), 
          conf.getInt("eventManager.handler.count", 10), false, conf);
      this.server.start();
      Thread eventChannelThread = new Thread(new EventChannel(), "Event Channel Thread");
      Thread eventEntryStoreThread = new Thread(new EventEntryMonitor(), "Event Entry Monitor Thread");
      Thread notificationChannelThread = new Thread(new NotificationChannel(), "Notification Channel Thread");
      eventChannelThread.start();
      eventEntryStoreThread.start();
      notificationChannelThread.start();
      addRunningThread(eventChannelThread);
      addRunningThread(eventEntryStoreThread);
      addRunningThread(notificationChannelThread);

    } catch (ChannelException e) {
      e.printStackTrace();
    } catch (UnknownHostException e) {
      e.printStackTrace();
    } catch (IOException e) {
      e.printStackTrace();
    }
  }
  
  private void addRunningThread(Thread t) {
    synchronized(runningThreads) {
      runningThreads.add(t);
    }
  }
  
  class EventChannel extends ReceiverAdapter implements Runnable {
    
    private String localAddressForEvent = "";
    public EventChannel() throws ChannelException {
      eventChannel = new JChannel(Constants.EVENT_ADDRESS + Constants.SOCK_PROPERTY);
      eventChannel.setReceiver(this);
      eventChannel.connect(eventChannelName);
    }

    public void run() {
      try {
        LOG.info("---------- EventChannel Thread Started from EventManager ----------");
        eventLoop();
      } catch (ChannelException e) {
        e.printStackTrace();
      }
    }

    private void eventLoop() throws ChannelException {

      while (!stopRequested.get()) {
        try {
          Thread.sleep(20);
        } catch (InterruptedException e) {
          eventChannel.close();
          eventChannel.shutdown();
          LOG.fatal("event server:" + localAddressForEvent + " is shutdown" );
          return;
        }
      }
      eventChannel.close();
      eventChannel.shutdown();
    }

    public void receive(Message msg) {

      Object message = msg.getObject();
      try {
        if (isPrimaryNode) {
          if (message instanceof EventMessage) {
            EventMessage eventMessage = (EventMessage) msg.getObject();
            synchronized(eventDeliveryQueue) {
              if (messageParser.isValidMessage(eventMessage)) {
                eventDeliveryQueue.add(eventMessage);  
                eventDeliveryQueue.notifyAll();
              } else {
                LOG.fatal("FATAL ERROR : MESSAGE FORMAT IS NOT VALID " + eventMessage);
              }
            }
          } else if (message instanceof EventMessage[]) {            
            EventMessage[] eventMessages = (EventMessage[]) msg.getObject();
            synchronized(eventDeliveryQueue) {
              for(EventMessage eventEntry : eventMessages) {
                if (messageParser.isValidMessage(eventEntry)) {
                  eventDeliveryQueue.add(eventEntry);  
                } else {
                  LOG.fatal("FATAL ERROR : MESSAGE FORMAT IS NOT VALID " + eventEntry);
                }
              }
              eventDeliveryQueue.notifyAll();
            }
          } else if (message instanceof MemberServerInfo) {
            LOG.debug("Received MemberServerInfo : "
                + ((MemberServerInfo) message).getServerAddress());
            membershipQueue.add((MemberServerInfo) message);
          } else if (message instanceof MasterServerInfo) {
            LOG.debug("Received MasterServerInfo by Primary Node");
            return;
          } else {
            LOG.error("Received unknown type of message");
            LOG.fatal(message.getClass().getCanonicalName());
          }
        } else {
          if (message instanceof MemberServerInfo) {
            LOG.debug("Received MemberServerInfo by Secondary Node");
            return;
          } else if (message instanceof MasterServerInfo) {
            masterEventManager = ((MasterServerInfo)message).getServerAddress();
          } else if(message instanceof EventMessage) {
            if(((EventMessage) message).getEventType() == LockService.Events.MASTER_FAILOVER.ordinal()) {
              sendEventToClient((EventMessage) message);
            }
          }
        }
      } catch (Exception e) {
        e.printStackTrace();
      }
    }

    public void viewAccepted(View view) {
      String creator = view.getCreator().toString();
      LOG.info("Creator : -----" + creator + "-----");
      localAddressForEvent = eventChannel.getLocalAddress().toString();
      if (localAddressForEvent.equals("")) {
        while (true) {
          if (eventChannel.isConnected()) {
            localAddressForEvent = eventChannel.getLocalAddress().toString();
            if (!localAddressForEvent.equals("")) {
              break;
            }
          }
          sleeper.sleepMilliSeconds(50);
        }
      }
      if (localAddressForEvent.equals(creator)) {
        isPrimaryNode = true;
        masterEventManager = getHostName();
        LOG.info("<<<<<Primary : " + localAddressForEvent + " for Event Channel>>>>>");
        LOG.info("Primary Node's View : " + view);
      } else {
        isPrimaryNode = false;
        LOG.info("<<<<<Secondary : " + localAddressForEvent + " for Event Channel>>>>>");
      }
      
      new Thread() {
        public void run() {
          try {
            if(isPrimaryNode) {
              updateMembership();
            } 
            while(!eventChannel.isConnected()) {
              sleeper.sleepMilliSeconds(10);
            }
            if(isPrimaryNode) {
              eventChannel.send(new Message(null, null, new MasterServerInfo(getHostName())));
            } else {
              MemberServerInfo mServerInfo = new MemberServerInfo(getHostName());
              mServerInfo.setMaster(false);
              eventChannel.send(new Message(eventChannel.getView().getCreator(), null, mServerInfo));
              LOG.debug("send slave event manager : " + getHostName() + " from event manager");  
            }
          } catch (ChannelException e) {
            e.printStackTrace();
          } 
        }
      }.start();      
    }    
    
    private void updateMembership() {      
      new Thread() {
        public void run() {                    
          MemberServerInfo[] copiedGroupMembers;
          try {
            sleeper.sleepMilliSeconds(500);
            numberOfEventManagers.set(eventChannel.getView().size());
            synchronized(membershipQueue) {
              while(membershipQueue.size() > numberOfEventManagers.get()) {
                membershipQueue.remove();
              }
              MemberServerInfo mServerInfo = new MemberServerInfo(getHostName());
              membershipQueue.add(mServerInfo);
              copiedGroupMembers = new MemberServerInfo[membershipQueue.size()];
              membershipQueue.toArray(copiedGroupMembers);
              boolean masterEventManagerIncluded = false;
              Set<String> memberSet = new HashSet<String>();
              for(MemberServerInfo member : copiedGroupMembers) {
                if(member.isMaster()) {
                  if(member.getServerAddress().equals(hostName)) {
                    if(masterEventManagerIncluded) {
                      membershipQueue.remove(member);
                    } else {
                      masterEventManagerIncluded = true;
                      memberSet.add(hostName);
                    }
                  } else {
                    membershipQueue.remove(member);
                  }
                } else {
                  if(memberSet.contains(member.getServerAddress())) {
                    membershipQueue.remove(member);
                  } else {
                    memberSet.add(member.getServerAddress());
                  }
                }
              }
              copiedGroupMembers = null;
              copiedGroupMembers = new MemberServerInfo[membershipQueue.size()];
              membershipQueue.toArray(copiedGroupMembers);
            }
            
            if (copiedGroupMembers.length > 0) {
              boolean channelOpened = true;
              int exitCount = 0;
              while (!eventChannel.isConnected()) {
                sleeper.sleepMilliSeconds(50);
                if (++exitCount > 10) {
                  channelOpened = false;
                }
              }
              if (channelOpened) {
                eventChannel.send(new Message(null, null, copiedGroupMembers));
              }
            }
          } catch (ChannelException e) {
            e.printStackTrace();
          } 
        }
      }.start();
    }
  }
  
  public class EventEntryMonitor implements Runnable {

    Map<String, LeaseTime> copiedEventEntryStore = null;
    long currentTime;
    
    public void run() {
      LOG.info("---------- EventEntryMonitor Thread Started from EventManager ----------");
      entryMonitorLoop();
    }

    private void entryMonitorLoop() {
      while (!stopRequested.get()) {
        try {
          if (isPrimaryNode) {
            copiedEventEntryStore = new ConcurrentHashMap<String, LeaseTime>();
            copiedEventEntryStore.putAll(eventEntryStore);
            for (Map.Entry<String, LeaseTime> entry : copiedEventEntryStore.entrySet()) {
              currentTime = System.currentTimeMillis();
              if( (currentTime - entry.getValue().getRegisterTime()) > entry.getValue().getLeaseTime()) {
                eventEntryStore.remove(entry.getKey());
              }
            }
            copiedEventEntryStore.clear();
          }
          Thread.sleep(pollingInterval);
        } catch (InterruptedException e) {
          eventEntryStore.stop();
          return;
        }
      }
    }
    
  }
  
  private void sendEventToClient(EventMessage eventMessage) throws IOException {
    LOG.info(eventMessage + " from Event Manager");
    boolean isMessageSent = false;
    int loopCount = 0;
    while (!isMessageSent) {
      if (loopCount > 10) {
        LOG.info("Problem with notificationChannel and dropped lock event " + eventMessage);
        break;
      }
      try {
        if (notificationChannel == null || !notificationChannel.isConnected()) {
          LOG.debug("notificationChannel is not open : can not send LockEvent message");
          isMessageSent = false;
        } else {
          LOG.debug("LockEvent: " + eventMessage + " from Event Manager");
          notificationChannel.send(new Message(null, null, eventMessage));
          isMessageSent = true;
          break;
        }
        ++loopCount;
        sleeper.sleepMilliSeconds(100);
      } catch (ChannelException e) {
        throw new IOException(e.getMessage());
      } 
    }
  }
  
  private void sendEventToClient(List<SpaceMessage> spaceMessages) throws IOException {
    boolean isMessageSent = false;
    int loopCount = 0;
    while (!isMessageSent) {
      if (loopCount > 10) {
        LOG.debug("Problem with notificationChannel and dropped space event");
        break;
      }
      try {
        if (notificationChannel == null || !notificationChannel.isConnected()) {
          LOG.debug("notificationChannel is not open : can not send SpaceEvent message");
        } else {
          for(SpaceMessage message : spaceMessages) {
            LOG.debug("SpaceEvent: " + message);
            notificationChannel.send(new Message(null, null, message)); 
          }
          isMessageSent = true;
          return;
        }
        sleeper.sleepMilliSeconds(500);
      } catch (ChannelException e) {
        throw new IOException(e.getMessage());
      } 
    }
  }
  
  class NotificationChannel extends ReceiverAdapter implements Runnable {

    public NotificationChannel() throws ChannelException, IOException {
      
      if(conf.getBoolean("event.delivery.multicast", true) ) {
        notificationChannel = new JChannel(Constants.NOTIFICATION_ADDRESS + Constants.SOCK_PROPERTY);
        
      } else {
        String localIpAddress = "";
        String property = "";
        try {
          String localHostName = InetAddress.getLocalHost().getCanonicalHostName();
          localIpAddress = InetAddress.getByName(localHostName).getHostAddress();
        } catch (IOException e) {
          e.printStackTrace();
        }
        String[] allPleiades = conf.getStrings("pleiades.cell.members");
        if (allPleiades == null || allPleiades.length == 0) {
          LOG.error("Can't get pleiades cell info check pleiades.cell.members option in pleiades-site.xml");
          throw new IOException(
              "Can't get pleiades cell info check pleiades.cell.member option in pleiades-site.xml");
        }
        String pleiadesHostNames = "";
        for (String entry : allPleiades) {
          pleiadesHostNames += (entry + "[7800],");
        }
        if (pleiadesHostNames.endsWith(",")) {
          pleiadesHostNames = pleiadesHostNames.substring(0, pleiadesHostNames.length() - 1);
          property = Constants.TCP_NIO_PROPERTY.replaceFirst(
              "hosts=localhost\\[7800\\]", "hosts=" + pleiadesHostNames);
        } else {
          property = Constants.TCP_NIO_PROPERTY;
        }

        if (localIpAddress != null && !localIpAddress.equals("")
            && !localIpAddress.equals("127.0.0.1")) {
          property = property.replaceAll("localhost", localIpAddress);
        } else {
          LOG.info("failed to get real ip address and could cause join operation to fail");
        }

        notificationChannel = new JChannel(property);
      }  
      notificationChannel.setReceiver(this); 
      notificationChannel.connect(getClusterID()
          + conf.get("notificationChannelName", "com.nhncorp.neptune.notificationChannel"));
    }
    
    public void run() {
      LOG.info("--------------- NotificationChannel Thread Started from EventManager---------------");
      notificationChannelLoop();
    }

    private void notificationChannelLoop() {
      while (!stopRequested.get()) {
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          notificationChannel.close();
          notificationChannel.shutdown();
          break;
        }
      }
      notificationChannel.close();
      notificationChannel.shutdown();
    }

    public void receive(Message msg) {
    }
    
    public void viewAccepted(View view) {
    }
  }
  
  private void startupEventDeliver() throws Throwable {
    
    int handlerCount = conf.getInt("pleiades.lockEventHandler.count", 3);
    eventDeliverList = new ArrayList<LockEventDeliver>();
    for(int count=0; count < handlerCount; ++count) {
      LockEventDeliver eventDeliver = new LockEventDeliver("event deliver thread " + String.valueOf(count));
      eventDeliver.start();
      eventDeliverList.add( eventDeliver );
    }
  }   
  
  class LockEventDeliver extends Thread {
    public LockEventDeliver(String name) {
      super(name);
    }
    public void run() {
      while (!stopRequested.get()) {
        try {
          EventMessage[] events = null;
          synchronized (eventDeliveryQueue) {
            while (eventDeliveryQueue.isEmpty()) {
              eventDeliveryQueue.wait();
            }
            events = eventDeliveryQueue.toArray(new EventMessage[]{});
            eventDeliveryQueue.clear();
          }
          for(EventMessage eventEntry : events) {
            sendEventToClient(eventEntry);  
          }
        } catch (InterruptedException e) {
          return;
        } catch (IOException e) {
          e.printStackTrace();
          return;
        }
      }
    }
  }  
  
  
  
  public String getHostName() {
    return this.hostName;
  }
  
  public void shutdown() {
    LOG.debug("shutdown eventManager");
    stopRequested.set(true);
    synchronized (runningThreads) {
      for (Thread thread : runningThreads) {
        thread.interrupt();
      }
    }
    sleeper.sleep();
    server.stop();
  }
  
  public String getClusterID() {
    return this.clusterID;
  }
  
  public boolean drop(EventMetaData metaData) {
    return false;
  }

  public boolean register(EventMetaData metaData, long timeout) {
    boolean isSuccess = false; 
    if (isPrimaryNode) {
      try {
        eventEntryStore.put(metaData.toString(), new LeaseTime(metaData.getLeaseTime()));
        isSuccess = true;  
        
      } catch (Exception e) {
      } finally {
        if(!isSuccess)
          return isSuccess;
      }
    }
    return isSuccess;
  }

  public boolean renewLease(EventMetaData metaData, long timeout) {
    return false;
  }
  
  public boolean read(EventMetaData metaData) {
    return false;
  }

  public boolean take(EventMetaData metaData) {
    return false;
  }

  public boolean write(EventMetaData metaData) {
    
    boolean isSuccess = false;
    boolean sendEventToClient = false;
    try {
      if(metaData.getEventType().equals("")) {
        return isSuccess;
      }
      if(eventSpace.containsKey(metaData.getEventType())) {
        EventMetaData[] oldMetaDatas = eventSpace.get(metaData.getEventType());
        EventMetaData[] newMetaDatas = new EventMetaData[oldMetaDatas.length+1];
        newMetaDatas[oldMetaDatas.length] = metaData;
        System.arraycopy(oldMetaDatas, 0, newMetaDatas, 0, oldMetaDatas.length);
        oldMetaDatas = null;
      } else {
        eventSpace.put(metaData.getEventType(), new EventMetaData[]{metaData});
      }
      
      
      
      
      if (eventEntryStore.containsKey(metaData)) {
        sendEventToClient = true;
      } else {
        Map<String, LeaseTime> copiedEventEntryStore = new HashMap<String, LeaseTime>();
        copiedEventEntryStore.putAll(eventEntryStore);
        for (Map.Entry<String, LeaseTime> entry : copiedEventEntryStore.entrySet()) {
          String serverType = entry.getKey().substring(0,
              entry.getKey().indexOf("::"));
          if (serverType.equals(metaData.getEventType())) {
            sendEventToClient = true;
            break;
          } else if (serverType.equals(metaData.getEventType() + ","
              + metaData.getEventKey())) {
            sendEventToClient = true;
            break;
          }
        }
        copiedEventEntryStore.clear();
        copiedEventEntryStore = null;
      }
      if(sendEventToClient) {
        List<SpaceMessage> spaceMessageList = new ArrayList<SpaceMessage>();
        spaceMessageList.add(new SpaceMessage(metaData.getEventType(), 
            metaData.getEventKey(), metaData.getOwnerId()));
        sendEventToClient(spaceMessageList);
      }
      isSuccess = true;

    } catch (Exception e) {
    } finally {
      if (!isSuccess)
        return isSuccess;
    }
    return isSuccess;

  }
  
  public boolean write(EventMetaData[] metaDatas) {
    
    boolean isSuccess = false;
    try {
      List<SpaceMessage> spaceMessageList = new ArrayList<SpaceMessage>();
      for(EventMetaData metaData : metaDatas) {
        if (eventEntryStore.containsKey(metaData)) {
          spaceMessageList.add(new SpaceMessage(metaData.getEventType(), 
              metaData.getEventKey(), metaData.getOwnerId()));
        } else {
          Map<String, LeaseTime> copiedEventEntryStore = new HashMap<String, LeaseTime>();
          copiedEventEntryStore.putAll(eventEntryStore);
          for (Map.Entry<String, LeaseTime> entry : copiedEventEntryStore.entrySet()) {
            String serverType = entry.getKey().substring(0, entry.getKey().indexOf("::"));
            if (serverType.equals(metaData.getEventType())) {
              spaceMessageList.add(new SpaceMessage(metaData.getEventType(), 
                  metaData.getEventKey(), metaData.getOwnerId()));
              break;
            } else if (serverType.equals(metaData.getEventType() + "," + metaData.getEventKey())) {
              spaceMessageList.add(new SpaceMessage(metaData.getEventType(), 
                  metaData.getEventKey(), metaData.getOwnerId()));
              break;
            }
          }
          copiedEventEntryStore.clear();
          copiedEventEntryStore = null;
        }
      }
      if(!spaceMessageList.isEmpty()) {
        sendEventToClient(spaceMessageList);
      }
      isSuccess = true;

    } catch (Exception e) {
    } finally {
      if (!isSuccess)
        return isSuccess;
    }
    return isSuccess;    
   }

  public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
    if (protocol.equals(EventServiceProtocol.class.getName())) {
      return EventServiceProtocol.versionID;
    } else {
      throw new IOException("Unknown protocol to cluster manager: " + protocol);
    }
  }
  
  public boolean isEventManagerReady() {
    return false;
  }
  
  public void run() {
    try {
      startupEventDeliver();
    } catch (Throwable e) {
      e.printStackTrace();
    }
    while (!isPrimaryNode) {
      sleeper.sleep();
      LOG.info("----- Event Manager is Waiting from : " + getHostName() + " -----");
      if (stopRequested.get())
        break;
    }
    doMainLoop();
  }
  
  private void doMainLoop() {
    while(!stopRequested.get()) {
      sleeper.sleepMilliSeconds(500);
    }
  }

  public static void main(String[] args) {
    EventManager eventManager = new EventManager(new PleiadesConfiguration());
    (new Thread(eventManager)).start();
  }

}
